/*
 *  TazioTransaction.java        22-ott-2009
 *   
 * Copyright (c) 2009 mentalsmash.org
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 * 
 *  http://www.apache.org/licenses/LICENSE-2.0
 * 
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 * 
 *  The use of the Apache License does not indicate that this project is
 *  affiliated with the Apache Software Foundation.
 */

package org.mentalsmash.tazio.transactions.management;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Future;

import org.mentalsmash.documentation.Concurrency;
import org.mentalsmash.tazio.commons.identifiers.OID;
import org.mentalsmash.tazio.transactions.RollbackException;
import org.mentalsmash.tazio.transactions.TransactionStatus;
import org.mentalsmash.tazio.transactions.TransactionalSystemException;
import org.mentalsmash.tazio.transactions.TransactionalSystemRuntimeException;
import org.mentalsmash.tazio.transactions.identifiers.TID;
import org.mentalsmash.tazio.transactions.objects.Snapshooter;
import org.mentalsmash.tazio.transactions.objects.Snapshot;
import org.mentalsmash.tazio.transactions.objects.TransactionalObject;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import com.google.common.base.Preconditions;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
import com.google.common.collect.Sets;
import com.google.inject.Inject;

/**
 * Base implementation of TazioTransactions. Actually it quite does all the
 * necessary stuff for a Tazio Transaction
 * 
 * @version 0.0.1 22/ott/2009
 * @author Andrea Reale <andrea.reale@gmail.com>
 */
@Concurrency(concurrentAccess = true, specification = "a TazioTransactionImpl instance, is inherently associated to a thread, "
	+ "and this is done at creation time by its factory that is the "
	+ "TransactionManager. Nevertheless an instance may be - during the "
	+ "commitring phase - also accessed by a committer thread: I need to make "
	+ "sure that while it is committing the status of a transaction is not "
	+ "modified by the thread owning it.")
abstract class TazioTransactionImpl implements TazioTransaction {

    private static final long serialVersionUID = -1599381999311796144L;

    private transient final static Logger log = LoggerFactory
	    .getLogger(TazioTransactionImpl.class);

    private final TID transactionID;

    private long seqNo = INITIAL_SEQNO; // initialized to an invalid seqno

    private TransactionStatus status = TransactionStatus.NO_TRANSACTION;

    // snapshot are only used locally to achieve isolation in transactions
    private final transient Map<OID, Snapshot> snapshots = Maps.newHashMap();

    // map of operations performed on transactional objects by this transaction.
    // It is sent to the RM cluster for commitment
    // along with the write set and read set
     private final Map<OID, List<TransactionalOperation>> operations = Maps
    	    .newHashMap();
    private final List<TransactionalOperation> writeOps = Lists.newLinkedList();
    private final List<TransactionalOperation>  readOps = Lists.newLinkedList();
    
    private final Set<OID> writeSet = Sets.newHashSet();
    private final Set<OID> readSet = Sets.newHashSet();

    // internally synchronized. Remember to synchronize over iterators
    private transient Set<TxStateObserver> stateObservers;

    // responsible for taking snapshot of transactional objects
    private final transient Snapshooter snapshooter;

    // if true the given transaction can do nothing then rollback
    private boolean rollBackOnly = false;

    @Inject
    protected TazioTransactionImpl(final SystemTransactionManager tm,
	    final Snapshooter snapshooter, final TID tid) {
	final Set<TxStateObserver> unsyncObservers = Sets.newHashSet();
	this.stateObservers = Collections.synchronizedSet(unsyncObservers);
	this.transactionID = tid;
	this.snapshooter = snapshooter;

    }

    /**
     * @return the stateObservers
     */
    private Set<TxStateObserver> getStateObservers() {
	synchronized (this) {

	    if (stateObservers == null) { // this may happen afer serialization
		final Set<TxStateObserver> unsyncObservers = Sets.newHashSet();
		stateObservers = Collections.synchronizedSet(unsyncObservers);
	    }
	}
	return stateObservers;
    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transactions.control.impl.TazioTransaction#
     * registerStateObserver
     * (org.mentalsmash.tazio.transactions.control.TxStateObserver)
     */
    public void registerStateObserver(final TxStateObserver observer) {
	log.trace("Adding a state observer for tx {} : {}", this.transactionID,
		observer);

	this.getStateObservers().add(observer);
    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transactions.control.impl.TazioTransaction#
     * removeStateObserver
     * (org.mentalsmash.tazio.transactions.control.TxStateObserver)
     */
    public boolean removeStateObserver(final TxStateObserver observer) {
	log.trace("Removing a state obeserver for tx {} : {}",
		this.transactionID, observer);
	return this.stateObservers.remove(observer);
    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transactions.control.impl.TazioTransaction#
     * getTransactionId()
     */
    @Override
    public TID getTransactionId() {
	return this.transactionID;
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.impl.TazioTransaction#getSeqNo
     * ()
     */
    public synchronized long getSeqNo() {
	return this.seqNo;
    }

    /**
     * Sets the sequence number of this transaction to the given value
     * 
     * @param seqNo
     *            the new sequece number
     * @throws IllegalSeqNoException
     *             if the provided sequence number is negative
     */
    @Override
    public synchronized void setSeqNo(final long seqNo) {
	if (seqNo < 0) {
	    log.error("Someone tried to set transaction's seqno to {}", seqNo);
	    throw new IllegalSeqNoException(
		    "Someone tried to set transaction's seqno to " + seqNo);
	}
	this.seqNo = seqNo;
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.impl.TazioTransaction#getStatus
     * ()
     */
    @Override
    public synchronized TransactionStatus getStatus() {
	return this.status;
    }

    /**
     * Changes the internal status of the <tt>TazioTransaction</tt>. This method
     * does not make a complete state transitions, but simply sets the internal
     * variable.
     * <p>
     * <strong>Consider</strong> using
     * {@link #stateTransition(TransactionStatus)} instead
     * </p>
     * 
     * @param newStatus
     *            the new status to set
     * @see #stateTransition(TransactionStatus)
     */
    protected synchronized void setStatus(final TransactionStatus newStatus) {
	this.status = newStatus;
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.impl.TazioTransaction#isActive
     * ()
     */
    public synchronized boolean isActive() {
	final boolean notActive = this.getStatus().equals(
		TransactionStatus.NO_TRANSACTION)
		|| this.getStatus().equals(TransactionStatus.COMMITTED)
		|| this.getStatus().equals(TransactionStatus.ROLLED_BACK);
	return !notActive;
    }

    /**
     * Changes the state of the <tt>TazioTransaction</tt> to the given
     * {@link TransactionStatus} and notifies the registered
     * {@link TxStateObserver}s
     * 
     * @param to
     *            the destination status to switch to
     */
    @Override
    public void stateTransition(final TransactionStatus to) {
	if (!to.equals(getStatus())) { // avoids useless state transitions and
	    // notifications
	    final TransactionStatus from = this.status;
	    setStatus(to);
	    synchronized (this.getStateObservers()) {
		for (final TxStateObserver observer : this.getStateObservers()) {
		    observer.statusChanged(from, to, this.getTransactionId());

		}
	    }
	}
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.impl.TazioTransaction#getReadSet
     * ()
     */
    public synchronized Set<OID> getReadSet() {
	return Collections.unmodifiableSet(this.readSet);
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.impl.TazioTransaction#getWriteSet
     * ()
     */
    public synchronized Set<OID> getWriteSet() {
	return Collections.unmodifiableSet(this.writeSet);
    }

    /**
     * Adds an oid to the write set, indicating that the corresponding object
     * was written in the context of this transaction
     * 
     * @param oid
     *            the {@link OID} of the object to add
     */
    @Override
    public synchronized void addToWriteSet(final OID oid) {
	this.writeSet.add(oid);
    }

    /**
     * Adds an oid to the read set, indicating that the corresponding object was
     * read in the context of this transaction
     * 
     * @param oid
     *            the {@link OID} of the object to add
     */
    @Override
    public synchronized void addToReadSet(final OID oid) {
	this.readSet.add(oid);
    }

    /**
     * Returns the private snapshot for a transactional object, or
     * <code>null</code> if there is no snapshot for such an object
     * 
     * @param objectId
     *            the {@link OID} instance identifying a transactional object
     * @return a private snapshot of the object identified by
     *         <code>objectId</code> or <code>null</cod> if no such an object
     */
    @Override
    public synchronized Snapshot retrieveSnapshot(final OID objectId) {
	return this.snapshots.get(objectId);
    }

    /**
     * Returns an unmodifiable view of a map containing the snapshot taken by
     * this transaction
     * 
     * @return an unmodifiable view of a map containing the snapshot taken by
     *         this transaction
     */
    protected synchronized Map<OID, Snapshot> getSnapshots() {
	return Collections.unmodifiableMap(this.snapshots);
    }

    /**
     * Returns the list of operations performed by this transaction on a given
     * transactional object, or <code>null</code> if no operation was performed
     * on the given object
     * <p>
     * <strong>BEWARE:</strong> the methods returns <code>null</code> even if no
     * object having the desired {@link OID} exists
     * </p>
     * 
     * @return he list of operations performed by this transaction on a given
     *         transactional object, or <code>null</code> if no operation was
     *         performed on the given object
     */
    @Override
    public synchronized List<TransactionalOperation> getOperations(
	    final OID objectId) {
	
	List<TransactionalOperation> list = this.operations.get(objectId);
	
	if (list == null) list = new ArrayList<TransactionalOperation>();
	
	return list; 
    }

    /**
     * returns the list of every write operation performed by this transaction.
     * <p>
     * Changing this map, <strong>does affect<strong> the state of the
     * transaction.
     * </p>
     * 
     * @return
     */
    @Override
    public synchronized List<TransactionalOperation> getWriteOperations() {
	return Collections.unmodifiableList(this.writeOps);
    }
    
    /**
     * returns the list of every read operation performed by this transaction.
     * <p>
     * Changing this map, <strong>does affect<strong> the state of the
     * transaction.
     * </p>
     * 
     * @return
     */
    @Override
    public synchronized List<TransactionalOperation> getReadOperations() {
	return Collections.unmodifiableList(this.readOps);
    }

    /**
     * Log an operation on the transaction
     * 
     * @param op
     *            the operation instance to log
     */
    @Override
    public void logOperation(final TransactionalOperation op) {
	 final OID objectId = op.getTargetObject();
	 //ADDING Both on the lists and the map to support fast retrieval by oid and oderede
	 //retrieval of ops
	synchronized (this) {
	    List<TransactionalOperation> operationsOnObj = this.operations
		    .get(objectId);
	    if (operationsOnObj == null) {
		operationsOnObj = Lists.newArrayList();
		this.operations.put(objectId, operationsOnObj);
	    }
	   
	    operationsOnObj.add(op);
	    if(op.isWriteOperation()){
		this.writeOps.add(op);
	    } else {
		this.readOps.add(op);
	    }
	}
    }

    /**
     * Takes a snapshot of the given {@link TransactionalObject} and adds it to
     * the private snapshots of this transaction
     * 
     * @param obj
     *            the {@link TransactionalObject} instance to take the snapshot
     *            of
     */
    @Override
    public synchronized void takeSnapshot(final TransactionalObject obj) {
	this.snapshooter.createSnapshot(obj, this.snapshots);
    }

    /* ************************************************************************ */
    /* From org.mentalsmash.tazio.transactions.Transaction interface */
    /* ************************************************************************ */

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.impl.TazioTransaction#commit()
     */
    @Override
    public void commit() throws RollbackException, TransactionStateException,
	    TransactionalSystemException {
	if (!this.getStatus().equals(TransactionStatus.RUNNING)) {
	    throw new TransactionStateException(
		    "Cannot commit a not running transaction");
	}

	if (!threadOwnsTransaction()) {
	    log
		    .error("A thread has tried to commit a transaction it does not own");
	    throw new TransactionalSystemRuntimeException(
		    "Thread does not own the transaction");
	}
	TransactionsStore.getSystemTransactionManager().commit();
    }

    /**
     * Checks whether this instance of <tt>TazioTransaction</tt> is the same
     * transaction that is currently associated to the running thread
     * 
     * @return <code>true</code> if this instance is the one associated to the
     *         current thread
     */
    private boolean threadOwnsTransaction() {
	return this.equals(TransactionsStore.getSystemTransactionManager()
		.getTransaction());
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.impl.TazioTransaction#rollback
     * ()
     */
    @Override
    public void rollback() throws TransactionStateException {
	Preconditions.checkState(threadOwnsTransaction(),
		"Thread does not own the transaction");
	Preconditions.checkState(this.getStatus().equals(
		TransactionStatus.RUNNING),
		"Cannot rollback a not running transaction");

	TransactionsStore.getSystemTransactionManager().rollback();
    }

    /*
     * (non-Javadoc)
     * 
     * @seeorg.mentalsmash.tazio.transactions.control.impl.TazioTransaction#
     * setRollbackOnly()
     */
    @Override
    public void setRollbackOnly() {
	Preconditions.checkState(threadOwnsTransaction(),
		"Calling thread must own the transaction");
	this.rollBackOnly = true;
    }

    /*
     * (non-Javadoc)
     * 
     * @see
     * org.mentalsmash.tazio.transactions.control.impl.TazioTransaction#commitAsync
     * ()
     */
    @Override
    public Future<TransactionStatus> commitAsync()
	    throws TransactionStateException {
	Preconditions.checkState(threadOwnsTransaction(),
		"A thread has tried to commit a transaction it does not own");

	return TransactionsStore.getSystemTransactionManager().commitAsync();
    }

    /**
     * Checks whether rollback is set to be the only possible outcome of this
     * transaction
     * 
     * @return <code>true</code> if this transaction has to rollback no matter
     *         what
     */
    @Override
    public boolean isRollbackOnly() {
	return this.rollBackOnly;
    }

    @Override
    public String toString() {
	return "TazioTransaction [seqNo=" + this.seqNo + ", transactionID="
		+ this.transactionID + "]";
    }

    @Override
    public boolean equals(final Object obj) {
	if (this == obj) {
	    return true;
	}
	if (obj == null) {
	    return false;
	}
	if (getClass() != obj.getClass()) {
	    return false;
	}
	final TazioTransactionImpl other = (TazioTransactionImpl) obj;
	if (this.getTransactionId() == null) {
	    if (other.getTransactionId() != null) {
		return false;
	    }
	} else if (!this.getTransactionId().equals(other.getTransactionId())) {
	    return false;
	}
	return true;
    }

    @Override
    public int hashCode() {
	final int prime = 3121;
	final int result = 1;
	return prime
		* result
		+ ((this.getTransactionId() == null) ? 0 : this
			.getTransactionId().hashCode());
    }

}
